package james.gan.remoting.netty.server;

import java.lang.reflect.Method;
import java.net.SocketAddress;
import java.util.List;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.DefaultChannelPromise;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.MessageToByteEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import james.gan.common.Request;
import james.gan.common.Response;
import james.gan.container.ServiceContainer;
import james.gan.remoting.netty.pool.NettyClientProxy;
import james.gan.serialization.api.Serialization;

public class NettyServerProxy {

	private final Logger logger = LoggerFactory.getLogger(NettyClientProxy.class);
	private final SocketAddress localAddress;
	private final Serialization serialization;
	private final ServiceContainer container;
	private EventLoopGroup boss;
	private EventLoopGroup worker;
	private Channel channel;

	public NettyServerProxy(SocketAddress localAddress, Serialization serialization, ServiceContainer container)
			throws InterruptedException {
		this.localAddress = localAddress;
		this.serialization = serialization;
		this.container = container;
		init();
	}

	public void init() throws InterruptedException {
		EventLoopGroup boss = new NioEventLoopGroup(1);
		EventLoopGroup worker = new NioEventLoopGroup();
		ChannelFuture future = new ServerBootstrap().group(boss, worker).localAddress(localAddress)
				.handler(new LoggingHandler(LogLevel.INFO)).childOption(ChannelOption.TCP_NODELAY, true)
				.childOption(ChannelOption.SO_KEEPALIVE, true).channel(NioServerSocketChannel.class)
				.childHandler(new ChannelInitializer<SocketChannel>() {

					@Override
					protected void initChannel(SocketChannel ch) throws Exception {

						ch.pipeline().addLast(new IdleStateHandler(2, 0, 0, TimeUnit.MINUTES))
								.addLast(new NettyDecoder(Request.class)).addLast(new NettyEncoder())
								.addLast(new NettyServerHandler());
					}
				}).bind();
		channel = future.sync().addListener(f -> {
			if (f.isSuccess()) {
				logger.info("Server channel: {} started", localAddress);
			}
		}).channel();

	}

	class NettyDecoder extends ByteToMessageDecoder {

		private Class<?> clzz;

		public NettyDecoder(Class<?> clzz) {
			this.clzz = clzz;
		}

		@Override
		protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
			if (in.readableBytes() < 4) {
				return;
			}
			in.markReaderIndex();
			int dataLength = in.readInt();
			if (dataLength < 0) {
				ctx.close();
			}
			if (in.readableBytes() < dataLength) {
				in.resetReaderIndex();
				return; // fix 1024k buffer splice limix
			}
			byte[] data = new byte[dataLength];
			in.readBytes(data);

			Object obj = serialization.deserialize(data, clzz);
			out.add(obj);
		}

	}

	class NettyEncoder extends MessageToByteEncoder<Response> {

		@Override
		protected void encode(ChannelHandlerContext ctx, Response msg, ByteBuf out) throws Exception {
			byte[] data = serialization.serialize(msg);
			out.writeInt(data.length);
			out.writeBytes(data);
		}
	}

	@ChannelHandler.Sharable
	class NettyServerHandler extends ChannelInboundHandlerAdapter {

		private int loss_connect_time = 0;

		@Override
		public void channelActive(ChannelHandlerContext ctx) throws Exception {
			loss_connect_time = 0;
			super.channelActive(ctx);
		}

		@Override
		public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
			Response response = new Response();
			if (msg instanceof Request) {
				Request req = (Request) msg;
				response.setId(req.getId());
				Object service = container.getBean(req.getClassName());
				if (service != null) {
					Method method = service.getClass().getMethod(req.getMethodName(), req.getParameterTypes());
					Object result = method.invoke(service, req.getParameters());
					response.setResult(result);
				}
				ctx.writeAndFlush(response, new DefaultChannelPromise(ctx.channel()).addListener(f -> {
					logger.debug("RE:{}", req.getId());
				}));
			} else {
				logger.debug("No support type:", msg.getClass());
				ctx.writeAndFlush(response);
			}
		}

		@Override
		public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
			if (evt instanceof IdleStateEvent) {
				IdleStateEvent event = (IdleStateEvent) evt;
				if (event.state() == IdleState.READER_IDLE) {
					loss_connect_time++;
					if (loss_connect_time > 2) {
						System.out.println("Close:" + ctx.channel().id().asShortText());
						ctx.channel().close();
					}
				}
			} else {
				super.userEventTriggered(ctx, evt);
			}
		}
	}

	public void close() {
		boss.shutdownGracefully();
		worker.shutdownGracefully();
		if (channel != null && channel.isOpen()) {
			channel.close();
		}
	}
}
